package com.gator.rabbitmq.pablo_rabbitmq.worker;


import com.gator.rabbitmq.pablo_rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Author PABLO
 * @Date 2022/6/4 22:11
 * @Desc 一个队列中的一个消息只能被处理一次
 */
public class WorkerProducer {
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) {
        //sendMsgOne();
        //sendMsgBatch();
        sendMsgAsync();
    }

    /**
     * @Description: 生产者发布消息单个确认
     * @Author: PABLO
     * @Date: 2022/6/5 11:42
     * @Params: []
     * @Return: void
     **/
    public static void sendMsgOne() {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //针对此通道开启发布确认，默认关闭
            channel.confirmSelect();
            //队列声明
            //durable 是否持久化
            channel.queueDeclare(QUEUE_NAME,
                    true,
                    false,
                    false,
                    null);
            long start = new Date().getTime();
            for (int i = 1; i <= 1000; i++) {
                String msg = i + "";
                channel.basicPublish("",
                        QUEUE_NAME,
                        //props 可设置消息属性为持久化
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        msg.getBytes(StandardCharsets.UTF_8));
                //发布确认
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息发布成功" + msg);
                }
            }
            long end = new Date().getTime();
            System.out.println("单个确认耗时" + (end - start) + "ms");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * @Description: 生产者发布消息批量确认
     * @Author: PABLO
     * @Date: 2022/6/5 11:42
     * @Params: []
     * @Return: void
     **/
    public static void sendMsgBatch() {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //针对此通道开启发布确认，默认关闭
            channel.confirmSelect();
            //队列声明
            //durable 是否持久化
            channel.queueDeclare(QUEUE_NAME,
                    true,
                    false,
                    false,
                    null);

            //批量个数
            int batchCount = 100;
            int num = 0;
            long start = new Date().getTime();
            for (int i = 1; i <= 1000; i++) {
                String msg = i + "";
                channel.basicPublish("",
                        QUEUE_NAME,
                        //props 可设置消息属性为持久化
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        msg.getBytes(StandardCharsets.UTF_8));
                if (i % batchCount == 0) {

                    System.out.println("第" + (++num) + "批次确认");
                    //发布确认
                    boolean flag = channel.waitForConfirms();
                    if (flag) {
                        System.out.println("消息发布成功" + msg);
                    }
                }
            }
            long end = new Date().getTime();
            System.out.println("批量确认耗时" + (end - start) + "ms");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * @Description: 生产者发布消息异步确认
     * @Author: PABLO
     * @Date: 2022/6/5 11:42
     * @Params: []
     * @Return: void
     **/
    public static void sendMsgAsync() {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //针对此通道开启发布确认，默认关闭
            channel.confirmSelect();
            //队列声明
            //durable 是否持久化
            channel.queueDeclare(QUEUE_NAME,
                    true,
                    false,
                    false,
                    null);

            //创建并发map,保存发送的消息，建立序号和信息的关联
            //key是long，表示消息的序号
            //value是具体消息类型
            ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();
            //定义成功的监听
            ConfirmCallback confirmSuccessCallback = (deliveryTag, multiple) -> {
                //multiple表示是否批量
                if (multiple) {
                    ConcurrentNavigableMap<Long, String> confirmed = map.headMap(deliveryTag);
                    confirmed.clear();
                } else {
                    //单个确认
                    map.remove(deliveryTag);
                }

                System.out.println("确认的消息" + deliveryTag);
            };
            //定义失败的监听
            ConfirmCallback confirmFailCallback = (deliveryTag, multiple) -> {
                //处理异步未确认消息
                String unConfirmed = map.get(deliveryTag);
                System.out.println("未确认的消息" + unConfirmed);
            };
            //准备监听器，监听broker回调
            channel.addConfirmListener(confirmSuccessCallback, confirmFailCallback);
            long start = new Date().getTime();
            for (int i = 1; i <= 1000; i++) {
                //标记发送消息
                String msg = i + "";
                //当处于确认模式时，返回下一个要发布的消息的序列号。
                map.put(channel.getNextPublishSeqNo(), msg);
                channel.basicPublish("",
                        QUEUE_NAME,
                        //props 可设置消息属性为持久化
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        msg.getBytes(StandardCharsets.UTF_8));
            }
            long end = new Date().getTime();
            System.out.println("异步确认耗时" + (end - start) + "ms");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
